package com.bwstudio.demo.provider.message;

import lombok.RequiredArgsConstructor;
import org.apache.rocketmq.spring.annotation.RocketMQTransactionListener;
import org.apache.rocketmq.spring.core.RocketMQLocalTransactionListener;
import org.apache.rocketmq.spring.core.RocketMQLocalTransactionState;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.Message;
import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.Transactional;

import com.alibaba.fastjson.JSONObject;
import com.bwstudio.demo.provider.dao.AccountInfoDao;
import com.bwstudio.demo.provider.model.AccountChangeEvent;
import com.bwstudio.demo.provider.service.AccountInfoService;

import lombok.extern.slf4j.Slf4j;

/**
 * @Description: 业务方法中执行 sendMessageInTransaction 时对应 txProducerGroup 的事务消息的监听器
 * txProducerGroup = "producer_group_txmsg_bank1" 这个group的作用范围在本应用的进程内，作为应用内部发送事务消息的监听目标
 * @Author: Bu Wei
 * @Date: 2020年3月11日 上午10:49:37
 */
@Component
@Slf4j
@RequiredArgsConstructor(onConstructor = @__(@Autowired))
//@RocketMQTransactionListener(txProducerGroup = "producer_group_txmsg_bank1")
@RocketMQTransactionListener
public class ProducerTxmsgListener implements RocketMQLocalTransactionListener {

    private final AccountInfoService accountInfoService;

	private final AccountInfoDao accountInfoDao;

	/**
	 * @Description: 事务消息发送后的回调方法，当消息发送给mq成功，此方法被回调
	 * @param message
	 * @param o
	 * @return
	 * @Author: Bu Wei
	 * @Date: 2020年3月11日 上午10:56:07
	 */
	@Override
	@Transactional
	public RocketMQLocalTransactionState executeLocalTransaction(Message message, Object o) {
		try {
			// 解析message，转成AccountChangeEvent
			String messageString = new String((byte[]) message.getPayload());
			JSONObject jsonObject = JSONObject.parseObject(messageString);
			String accountChangeString = jsonObject.getString("accountChange");
			// 将accountChange（json）转成AccountChangeEvent
			AccountChangeEvent accountChangeEvent = JSONObject.parseObject(accountChangeString,
					AccountChangeEvent.class);
			// 执行本地事务，扣减金额
			accountInfoService.doUpdateAccountBalance(accountChangeEvent);
			// 当返回RocketMQLocalTransactionState.COMMIT，自动向mq发送commit消息，mq将消息的状态改为可消费
			return RocketMQLocalTransactionState.COMMIT;
		} catch (Exception e) {
			e.printStackTrace();
			return RocketMQLocalTransactionState.ROLLBACK;
		}
	}

	/**
	 * @Description: 当一阶段half message发送成功，丢失了二阶段确认消息时，
	 * 进行半消息事务状态回查，查询本地事务是否执行成功（是否扣减金额）
	 * @param message
	 * @return
	 * @Author: Bu Wei
	 * @Date: 2020年3月11日 上午10:57:54
	 */
	@Override
	public RocketMQLocalTransactionState checkLocalTransaction(Message message) {
		// 解析message，转成AccountChangeEvent
		String messageString = new String((byte[]) message.getPayload());
		JSONObject jsonObject = JSONObject.parseObject(messageString);
		String accountChangeString = jsonObject.getString("accountChange");
		// 将accountChange（json）转成AccountChangeEvent
		AccountChangeEvent accountChangeEvent = JSONObject.parseObject(accountChangeString, AccountChangeEvent.class);
		// 事务id
		String txNo = accountChangeEvent.getTxNo();
		int existTx = accountInfoDao.isExistTx(txNo);
		if (existTx > 0) {
			return RocketMQLocalTransactionState.COMMIT;
		} else {
			return RocketMQLocalTransactionState.UNKNOWN;
		}
	}
}